package br.unifor.g2cl;



import java.io.IOException;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.logging.Level;
import java.util.logging.Logger;

import net.sf.jgcs.Annotation;
import net.sf.jgcs.ClosedSessionException;
import net.sf.jgcs.ControlSession;
import net.sf.jgcs.Message;
import net.sf.jgcs.Service;
import net.sf.jgcs.UnsupportedServiceException;
import net.sf.jgcs.membership.MembershipSession;

// FIXME problem: is it efficient to have one thread for each bundlesender? NO!!!!

public class BundleDataSession extends DataSessionAdapter {

	private static Logger logger = Logger.getLogger(BundleDataSession.class
			.getName());

	
	private static final int HEADER = (byte) 0xF;
	
	private static final int DEFAULT_MAX_SIZE = 100;
	private static final long DEFAULT_TIMEOUT = 20;

	private HashMap<SocketAddress, HashMap<Service, BundleSender>> map;
	private MembershipSession membershipSession;
	private int maxSize = DEFAULT_MAX_SIZE;
	private long timeout = DEFAULT_TIMEOUT;

	
	public BundleDataSession(IMarshalDataSession data, ControlSession control) {
		super(data);
		
		if (control instanceof MembershipSession) {
			this.membershipSession = (MembershipSession) control;
			// TODO add a listener. Delete destination from hashmap when a member quits the group
		}
		
		map = new HashMap<SocketAddress, HashMap<Service, BundleSender>>();

	}

	
	
	/* ********************************************************************** *
	 * **************************** Properties ****************************** *
	 * ********************************************************************** */
	
	public int getMaxSize() {
		return maxSize;
	}
	
	public void setMaxSize(String maxSize) {
		this.maxSize = Integer.parseInt(maxSize);
	}
	
	public long getTimeout() {
		return timeout;
	}
	
	public void setTimeout(String timeout) {
		this.timeout = Long.parseLong(timeout);
	}


	/* ********************************************************************** *
	 * ********** Overrided methods to add the new feature (bundling) ******* *
	 * ********************************************************************** */

	public void multicast(Message msg, Service service, Object cookie,
			Annotation... annotation) throws IOException,
			UnsupportedServiceException {

		HashMap<Service, BundleSender> m = map.get(null);
		
		if (m == null) {
			m = new HashMap<Service, BundleSender>();
			map.put(null, m);
		}

		if (!m.containsKey(service)) {
			BundleSender s = new BundleSender(service, null, msg.getSenderAddress());
			m.put(service, s);
		}
		BundleSender s = m.get(service);
		s.addMessage(msg.getPayload());
	}


	public void send(Message msg, Service service, Object cookie,
			SocketAddress destination, Annotation... annotation)
	throws IOException, UnsupportedServiceException {
		
		HashMap<Service, BundleSender> m = map.get(destination);
		
		if (m == null) {
			m = new HashMap<Service, BundleSender>();
			map.put(destination, m);
		}

		if (!m.containsKey(service)) {
			BundleSender s = new BundleSender(service, destination, msg.getSenderAddress());
			m.put(service, s);
		}
		BundleSender s = m.get(service);
		s.addMessage(msg.getPayload());
	}



	public Object onMessage(Message msg) {
		byte[] payload = msg.getPayload();
		int pos = 0;
		if (payload[0] != HEADER) {
			logger.severe("This message is not a bundled message");
			return null;
		}
		pos++;

		int numHeaders = Util.byteArrayToInt(payload, pos);
		pos += 4;

		for (int x = 0; x < numHeaders; x++) {
			int length = Util.byteArrayToInt(payload, pos);
			pos += 4;
			byte[] b = new byte[length];
			System.arraycopy(payload, pos, b, 0, b.length);
			pos += length;

			try {
				G2CLMessage g2 = (G2CLMessage) createMessage();
				g2.setSenderAddress(msg.getSenderAddress());
				g2.setPayload(b);
				realMessageListener.onMessage(g2);

			} catch (ClosedSessionException e) {
				logger.log(Level.SEVERE, "could not create the message", e);
			}
		}
		return null;
		
	}

	
	
	/* ********************************************************************** *
	 * ***************** Auxiliary classes for bundling ********************* *
	 * ********************************************************************** */

	private static Logger log = Logger.getLogger(BundleSender.class
			.getName());
	
	/**
	 * Represents a message to be sent (bundled)
	 */
	private class RawMessage {
		/** Counts the number of bytes the message has */
		private int counter;
		/** All messages that will be bundled into one single message */
		private List<byte[]> messages = new ArrayList<byte[]>();
		
		
		public byte[] getPayload() {
			byte[] payload = null;
			// inicio + num_msg + tam_msg1 + msg1 + tam_msg2 + msg2
			
			// creating final payload and filling header
			int numMessages = messages.size();
			// 1 byte for header, 4 for the number of messages and 4 bytes
			//for each message's size
			int s = counter + 1 + 4 + 4*numMessages;
			payload = new byte[s];
			payload[0] = HEADER;
			int pos = 1;
			Util.intToByteArray(payload, pos, numMessages);
			pos += 4;

			if (log.isLoggable(Level.FINER)) {
				log.finer("bundling message (" + numMessages + " messages)");
			}

			// putting each message in the buffer
			for (byte[] message : messages) {
				Util.intToByteArray(payload, pos, message.length);
				pos += 4;
				System.arraycopy(message, 0, payload, pos, message.length);
				pos += message.length;
			}
			return payload;
		}
		
	}
	

	private class BundleSender extends Thread {
		
		/** Task to send messages when timeout period expired */
		private TimerTask task;
		private Object lock = new Object();
		private RawMessage mess = new RawMessage();
		
		private SocketAddress dest;
		private Service service;
		private SocketAddress sender;
		private BlockingQueue<RawMessage> queue;

		public BundleSender(Service s, SocketAddress dest, SocketAddress sender) {
			this.service = s;
			this.sender = sender;
			this.dest = dest;
			this.queue = new LinkedBlockingQueue<RawMessage>();
			this.setName("Bundle Sender Thread");
			start(); // the thread
		}
		
		public void dispachar() {
			RawMessage m = mess;
			this.mess = new RawMessage(); // the magic goes here
			if (!queue.offer(m)) {
				throw new RuntimeException("pau dos brabo.... impossivel de acontecer");
			}
		}
		
		public void startNewJob() {
			if (log.isLoggable(Level.FINER)) {
				log.finer("starting new job");
			}

			Timer timer = new Timer();
			
			task = new TimerTask() {
				public void run() {
					synchronized (lock) {
						dispachar();
					}
				}
			};
			timer.schedule(task, timeout);
		}
		
		public void addMessage(byte[] message) {
			synchronized (lock) {
				if (mess.counter == 0) {
					startNewJob();
				}
				mess.messages.add(message);
				mess.counter += message.length;
				if (mess.counter >= maxSize) {
					if (task.cancel()) {
						dispachar();
					}
				}
			}
		}
		
		
		public void run() {
			while (true) {
				try {
					RawMessage m = queue.take();
					byte[] payload = m.getPayload();
					
					try {
						if (log.isLoggable(Level.FINER)) {
							log.finer("complete payload: " + Arrays.toString(payload));
						}
						G2CLMessage gm = (G2CLMessage) createMessage();
						gm.setPayload(payload);
						gm.setSenderAddress(sender);

						if (log.isLoggable(Level.FINER)) {
							log.finer("sending bundled message to " + dest);
						}

						if (dest == null) {
							data.multicast(gm, service, null, (Annotation[]) null);
						} else {
							data.send(gm, service, null, dest, (Annotation[]) null);
						}
					} catch (ClosedSessionException e) {
						logger.log(Level.SEVERE, "could not create the message", e);
					} catch (UnsupportedServiceException e) {
						logger.log(Level.SEVERE, "", e);
					} catch (IOException e) {
						logger.log(Level.SEVERE, "", e);
					}
					
				} catch (InterruptedException e) {
					logger.log(Level.SEVERE, "Error consuming messages from the queue", e);
				}
			}
		}		
	}
}

